/**
 * 授权给 Apache 软件基金会 (ASF) 一份
 * 或更多贡献者许可协议。 查看通知文件
 * 随本作品一起分发以获取更多信息
 * 关于版权所有权。 ASF 许可此文件
 * 根据 Apache 许可，版本 2.0（
 * “执照”）; 除非合规，否则您不得使用此文件
 * 带有许可证。 您可以在以下网址获取许可证副本
*
*     http://www.apache.org/licenses/LICENSE-2.0
*
 * 除非适用法律要求或书面同意，否则软件
 * 根据许可证分发是按“原样”分发的，
 * 没有任何形式的明示或暗示的保证或条件。
 * 请参阅许可证以了解管理权限的特定语言和
 * 许可下的限制。
*/

package org.apache.hadoop.mapred;

import static org.fusesource.leveldbjni.JniDBFactory.asString;
import static org.fusesource.leveldbjni.JniDBFactory.bytes;
import static org.jboss.netty.buffer.ChannelBuffers.wrappedBuffer;
import static org.jboss.netty.handler.codec.http.HttpHeaders.Names.CONTENT_TYPE;
import static org.jboss.netty.handler.codec.http.HttpMethod.GET;
import static org.jboss.netty.handler.codec.http.HttpResponseStatus.BAD_REQUEST;
import static org.jboss.netty.handler.codec.http.HttpResponseStatus.FORBIDDEN;
import static org.jboss.netty.handler.codec.http.HttpResponseStatus.INTERNAL_SERVER_ERROR;
import static org.jboss.netty.handler.codec.http.HttpResponseStatus.METHOD_NOT_ALLOWED;
import static org.jboss.netty.handler.codec.http.HttpResponseStatus.NOT_FOUND;
import static org.jboss.netty.handler.codec.http.HttpResponseStatus.OK;
import static org.jboss.netty.handler.codec.http.HttpResponseStatus.UNAUTHORIZED;
import static org.jboss.netty.handler.codec.http.HttpVersion.HTTP_1_1;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.net.InetSocketAddress;
import java.net.URL;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.regex.Pattern;

import javax.crypto.SecretKey;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DataInputByteBuffer;
import org.apache.hadoop.io.DataOutputBuffer;
import org.apache.hadoop.io.ReadaheadPool;
import org.apache.hadoop.io.SecureIOUtils;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.proto.ShuffleHandlerRecoveryProtos.JobShuffleInfoProto;
import org.apache.hadoop.mapreduce.MRConfig;
import org.apache.hadoop.mapreduce.security.SecureShuffleUtils;
import org.apache.hadoop.mapreduce.security.token.JobTokenIdentifier;
import org.apache.hadoop.mapreduce.security.token.JobTokenSecretManager;
import org.apache.hadoop.mapreduce.task.reduce.ShuffleHeader;
import org.apache.hadoop.metrics2.MetricsSystem;
import org.apache.hadoop.metrics2.annotation.Metric;
import org.apache.hadoop.metrics2.annotation.Metrics;
import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem;
import org.apache.hadoop.metrics2.lib.MutableCounterInt;
import org.apache.hadoop.metrics2.lib.MutableCounterLong;
import org.apache.hadoop.metrics2.lib.MutableGaugeInt;
import org.apache.hadoop.security.proto.SecurityProtos.TokenProto;
import org.apache.hadoop.security.ssl.SSLFactory;
import org.apache.hadoop.security.token.Token;
import org.apache.hadoop.util.DiskChecker;
import org.apache.hadoop.util.Shell;
import org.apache.hadoop.util.concurrent.HadoopExecutors;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.proto.YarnServerCommonProtos.VersionProto;
import org.apache.hadoop.yarn.server.api.ApplicationInitializationContext;
import org.apache.hadoop.yarn.server.api.ApplicationTerminationContext;
import org.apache.hadoop.yarn.server.api.AuxiliaryService;
import org.apache.hadoop.yarn.server.nodemanager.containermanager.localizer.ContainerLocalizer;
import org.apache.hadoop.yarn.server.records.Version;
import org.apache.hadoop.yarn.server.records.impl.pb.VersionPBImpl;
import org.apache.hadoop.yarn.server.utils.LeveldbIterator;
import org.fusesource.leveldbjni.JniDBFactory;
import org.fusesource.leveldbjni.internal.NativeDB;
import org.iq80.leveldb.DB;
import org.iq80.leveldb.DBException;
import org.iq80.leveldb.Options;
import org.jboss.netty.bootstrap.ServerBootstrap;
import org.jboss.netty.buffer.ChannelBuffers;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFactory;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelFutureListener;
import org.jboss.netty.channel.ChannelHandler;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelUpstreamHandler;
import org.jboss.netty.channel.group.ChannelGroup;
import org.jboss.netty.channel.group.DefaultChannelGroup;
import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;
import org.jboss.netty.handler.codec.frame.TooLongFrameException;
import org.jboss.netty.handler.codec.http.DefaultHttpResponse;
import org.jboss.netty.handler.codec.http.HttpChunkAggregator;
import org.jboss.netty.handler.codec.http.HttpRequest;
import org.jboss.netty.handler.codec.http.HttpRequestDecoder;
import org.jboss.netty.handler.codec.http.HttpResponse;
import org.jboss.netty.handler.codec.http.HttpResponseEncoder;
import org.jboss.netty.handler.codec.http.HttpResponseStatus;
import org.jboss.netty.handler.codec.http.QueryStringDecoder;
import org.jboss.netty.handler.ssl.SslHandler;
import org.jboss.netty.handler.stream.ChunkedWriteHandler;
import org.jboss.netty.handler.timeout.IdleState;
import org.jboss.netty.handler.timeout.IdleStateAwareChannelHandler;
import org.jboss.netty.handler.timeout.IdleStateEvent;
import org.jboss.netty.handler.timeout.IdleStateHandler;
import org.jboss.netty.util.CharsetUtil;
import org.jboss.netty.util.HashedWheelTimer;
import org.jboss.netty.util.Timer;
import org.eclipse.jetty.http.HttpHeader;
import org.slf4j.LoggerFactory;

import org.apache.hadoop.thirdparty.com.google.common.annotations.VisibleForTesting;
import org.apache.hadoop.thirdparty.com.google.common.base.Charsets;
import org.apache.hadoop.thirdparty.com.google.common.cache.CacheBuilder;
import org.apache.hadoop.thirdparty.com.google.common.cache.CacheLoader;
import org.apache.hadoop.thirdparty.com.google.common.cache.LoadingCache;
import org.apache.hadoop.thirdparty.com.google.common.cache.RemovalListener;
import org.apache.hadoop.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.hadoop.thirdparty.protobuf.ByteString;

public class ShuffleHandler extends AuxiliaryService {

  private static final org.slf4j.Logger LOG =
      LoggerFactory.getLogger(ShuffleHandler.class);//日志
  private static final org.slf4j.Logger AUDITLOG =
      LoggerFactory.getLogger(ShuffleHandler.class.getName()+".audit");//审计日志
  public static final String SHUFFLE_MANAGE_OS_CACHE = "mapreduce.shuffle.manage.os.cache";//shuffle 管理操作系统缓存
  public static final boolean DEFAULT_SHUFFLE_MANAGE_OS_CACHE = true;//默认shuffle管理操作系统缓存

  public static final String SHUFFLE_READAHEAD_BYTES = "mapreduce.shuffle.readahead.bytes";//shuffle预读字节数
  public static final int DEFAULT_SHUFFLE_READAHEAD_BYTES = 4 * 1024 * 1024;//默认shuffle预读字节数

  public static final String MAX_WEIGHT =
      "mapreduce.shuffle.pathcache.max-weight";//路径缓存
  public static final int DEFAULT_MAX_WEIGHT = 10 * 1024 * 1024;//默认路径缓存

  public static final String EXPIRE_AFTER_ACCESS_MINUTES =
      "mapreduce.shuffle.pathcache.expire-after-access-minutes";//访问多少分钟过期
  public static final int DEFAULT_EXPIRE_AFTER_ACCESS_MINUTES = 5;//默认访问5分钟过期

  public static final String CONCURRENCY_LEVEL =
      "mapreduce.shuffle.pathcache.concurrency-level";//并发级别
  public static final int DEFAULT_CONCURRENCY_LEVEL = 16;//默认并发级别：16

  // 识别与客户端关闭套接字相关的错误的模式
  private static final Pattern IGNORABLE_ERROR_MESSAGE = Pattern.compile(
      "^.*(?:connection.*reset|connection.*closed|broken.*pipe).*$",
      Pattern.CASE_INSENSITIVE);

  private static final String STATE_DB_NAME = "mapreduce_shuffle_state";//状态数据库名称
  private static final String STATE_DB_SCHEMA_VERSION_KEY = "shuffle-schema-version";//状态数据库架构版本密钥
  protected static final Version CURRENT_VERSION_INFO = 
      Version.newInstance(1, 0);//当前版本信息

  private static final String DATA_FILE_NAME = "file.out";//数据文件名
  private static final String INDEX_FILE_NAME = "file.out.index";//索引文件名

  public static final HttpResponseStatus TOO_MANY_REQ_STATUS =
      new HttpResponseStatus(429, "TOO MANY REQUESTS");//Http响应状态：请求过多
  // This should kept in sync with Fetcher.FETCH_RETRY_DELAY_DEFAULT
  public static final long FETCH_RETRY_DELAY = 1000L;//获取重试延迟
  public static final String RETRY_AFTER_HEADER = "Retry-After";

  private int port;
  private ChannelFactory selector;
  private final ChannelGroup accepted = new DefaultChannelGroup();
  protected HttpPipelineFactory pipelineFact;
  private int sslFileBufferSize;
  
  /**
   * shuffle 是否应该使用 posix_fadvise 调用来管理操作系统缓存发送文件
   */
  private boolean manageOsCache;
  private int readaheadLength;
  private int maxShuffleConnections;
  private int shuffleBufferSize;
  private boolean shuffleTransferToAllowed;
  private int maxSessionOpenFiles;
  private ReadaheadPool readaheadPool = ReadaheadPool.getInstance();//获取实例

  private Map<String,String> userRsrc;
  private JobTokenSecretManager secretManager;

  private DB stateDb = null;

  public static final String MAPREDUCE_SHUFFLE_SERVICEID =
      "mapreduce_shuffle";//MAPREDUCE_SHUFFLE 服务程序

  public static final String SHUFFLE_PORT_CONFIG_KEY = "mapreduce.shuffle.port";//shuffle端口配置key
  public static final int DEFAULT_SHUFFLE_PORT = 13562;//默认shuffle端口配置key：13562

  public static final String SHUFFLE_LISTEN_QUEUE_SIZE =
      "mapreduce.shuffle.listen.queue.size";//shuffle监听队列大小
  public static final int DEFAULT_SHUFFLE_LISTEN_QUEUE_SIZE = 128;//默认shuffle监听队列大小为128

  public static final String SHUFFLE_CONNECTION_KEEP_ALIVE_ENABLED =
      "mapreduce.shuffle.connection-keep-alive.enable";//是否对shuffle连接保持alive
  public static final boolean DEFAULT_SHUFFLE_CONNECTION_KEEP_ALIVE_ENABLED = false;

  public static final String SHUFFLE_CONNECTION_KEEP_ALIVE_TIME_OUT =
      "mapreduce.shuffle.connection-keep-alive.timeout";//shuffle连接保持alive超时时间
  public static final int DEFAULT_SHUFFLE_CONNECTION_KEEP_ALIVE_TIME_OUT = 5; //默认shuffle连接保持活动超时时间为5秒

  public static final String SHUFFLE_MAPOUTPUT_META_INFO_CACHE_SIZE =
      "mapreduce.shuffle.mapoutput-info.meta.cache.size";//shuffle映射输出元信息缓存大小
  public static final int DEFAULT_SHUFFLE_MAPOUTPUT_META_INFO_CACHE_SIZE = 1000;//默认shuffle映射输出元信息缓存大小为1000

  public static final String CONNECTION_CLOSE = "close";

  public static final String SUFFLE_SSL_FILE_BUFFER_SIZE_KEY =
    "mapreduce.shuffle.ssl.file.buffer.size";//shuffle SSL 文件缓冲区大小key
  public static final int DEFAULT_SUFFLE_SSL_FILE_BUFFER_SIZE = 60 * 1024;//默认shuffle SSL 文件缓冲区大小

  public static final String MAX_SHUFFLE_CONNECTIONS = "mapreduce.shuffle.max.connections";//最大shuffle连接
  public static final int DEFAULT_MAX_SHUFFLE_CONNECTIONS = 0; //默认最大shuffle连接

  public static final String MAX_SHUFFLE_THREADS = "mapreduce.shuffle.max.threads";//最大shuffle线程
  //0 表示 Netty 默认为 2 * 可用处理器数
  public static final int DEFAULT_MAX_SHUFFLE_THREADS = 0;
  
  public static final String SHUFFLE_BUFFER_SIZE = 
      "mapreduce.shuffle.transfer.buffer.size";//shuffle缓冲区大小
  public static final int DEFAULT_SHUFFLE_BUFFER_SIZE = 128 * 1024;//默认shuffle缓冲区大小
  
  public static final String  SHUFFLE_TRANSFERTO_ALLOWED = 
      "mapreduce.shuffle.transferTo.allowed";//允许shuffle转移
  public static final boolean DEFAULT_SHUFFLE_TRANSFERTO_ALLOWED = true;//默认允许shuffle转移为true
  public static final boolean WINDOWS_DEFAULT_SHUFFLE_TRANSFERTO_ALLOWED = 
      false;//系统默认允许shuffle转移为false
  private static final String TIMEOUT_HANDLER = "timeout";

  public static final String SHUFFLE_MAX_SESSION_OPEN_FILES =
      "mapreduce.shuffle.max.session-open-files";// 在shuffle期间单个GET请求可以同时打开的最大文件数
  public static final int DEFAULT_SHUFFLE_MAX_SESSION_OPEN_FILES = 3;

  boolean connectionKeepAliveEnabled = false;
  private int connectionKeepAliveTimeOut;
  private int mapOutputMetaInfoCacheSize;
  private Timer timer;

  //统计失败的和成功的shuffle输出的数量
  @Metrics(about="Shuffle output metrics", context="mapred")
  static class ShuffleMetrics implements ChannelFutureListener {
    @Metric("Shuffle output in bytes")//以字节为单位的shuffle输出
        MutableCounterLong shuffleOutputBytes;
    @Metric("# of failed shuffle outputs")//失败的shuffle输出
        MutableCounterInt shuffleOutputsFailed;
    @Metric("# of succeeeded shuffle outputs")//成功的shuffle输出
        MutableCounterInt shuffleOutputsOK;
    @Metric("# of current shuffle connections")//当前 shuffle 连接数
        MutableGaugeInt shuffleConnections;

    @Override
    public void operationComplete(ChannelFuture future) throws Exception {
      if (future.isSuccess()) {
        shuffleOutputsOK.incr();//成功的shuffle输出自增
      } else {
        shuffleOutputsFailed.incr();//失败的shuffle输出自增
      }
      shuffleConnections.decr();//当前shuffle连接数自减
    }
  }

  private final MetricsSystem ms;
  final ShuffleMetrics metrics;

  class ReduceMapFileCount implements ChannelFutureListener {

    private ReduceContext reduceContext;

    public ReduceMapFileCount(ReduceContext rc)
    {
      this.reduceContext = rc;
    }

    /**
     * ChannelFuture的作用是用来保存Channel异步操作的结果。
     * 当一个I/O操作开始的时候，一个新的future对象就会被创建。
     * 如果I/O操作以成功、失败或者被取消中的任何一种状态结束了，那么这个future将会被标记为已完成.
     */
    @Override
    public void operationComplete(ChannelFuture future) throws Exception {
      if (!future.isSuccess()) {
        future.getChannel().close();
        return;
      }
      int waitCount = this.reduceContext.getMapsToWait().decrementAndGet();//对正在等待的map数量进行递减及获取
      if (waitCount == 0)//如果只有1个正在等待的map
      {
        metrics.operationComplete(future);
        if (reduceContext.getKeepAlive()) {//若持续alive则设置超时
          ChannelPipeline pipeline = future.getChannel().getPipeline();
          TimeoutHandler timeoutHandler = (TimeoutHandler)pipeline.get(TIMEOUT_HANDLER);
          timeoutHandler.setEnabledTimeout(true);//设置启用超时
        } else {
          future.getChannel().close();
        }
      }
      else {//等待的map大于等于2个
        pipelineFact.getSHUFFLE().sendMap(reduceContext);
      }
    }
  }

  /**
   * 维护每个 messageReceived() Netty 上下文的参数。
   * 允许来自 operationComplete() 的 sendMapOutput 调用
   */
  private static class ReduceContext {

    private List<String> mapIds;
    private AtomicInteger mapsToWait;
    private AtomicInteger mapsToSend;
    private int reduceId;
    private ChannelHandlerContext ctx;
    private String user;
    private Map<String, Shuffle.MapOutputInfo> infoMap;
    private String jobId;
    private final boolean keepAlive;

    public ReduceContext(List<String> mapIds, int rId,
                         ChannelHandlerContext context, String usr,
                         Map<String, Shuffle.MapOutputInfo> mapOutputInfoMap,
                         String jobId, boolean keepAlive) {

      this.mapIds = mapIds;
      this.reduceId = rId;
      /**
       用于跟踪尚未完成的map输出数量的原子计数。
       多个 futureListeners 的 operationComplete() 可以异步递减这个值。它用于决定何时应该关闭通道。
      */
      this.mapsToWait = new AtomicInteger(mapIds.size());
      /**
       * 用于跟踪已发送的map输出数量的原子计数。
       * 多次调用 sendMap() 可以增加这个值异步。用于决定接下来应该发送哪个mapId。
      */
      this.mapsToSend = new AtomicInteger(0);
      this.ctx = context;
      this.user = usr;
      this.infoMap = mapOutputInfoMap;
      this.jobId = jobId;
      this.keepAlive = keepAlive;
    }

    public int getReduceId() {
      return reduceId;
    }

    public ChannelHandlerContext getCtx() {
      return ctx;
    }

    public String getUser() {
      return user;
    }

    public Map<String, Shuffle.MapOutputInfo> getInfoMap() {
      return infoMap;
    }

    public String getJobId() {
      return jobId;
    }

    public List<String> getMapIds() {
      return mapIds;
    }

    public AtomicInteger getMapsToSend() {
      return mapsToSend;
    }

    public AtomicInteger getMapsToWait() {
      return mapsToWait;
    }

    public boolean getKeepAlive() {
      return keepAlive;
    }
  }

  ShuffleHandler(MetricsSystem ms) {
    super(MAPREDUCE_SHUFFLE_SERVICEID);
    this.ms = ms;
    metrics = ms.register(new ShuffleMetrics());
  }

  public ShuffleHandler() {
    this(DefaultMetricsSystem.instance());
  }
  //序列化就是指把Java对象转换为字节序列的过程,反序列化就是指把字节序列恢复为Java对象的过程。
  /**
   * 将 shuffle 端口序列化为 ByteBuffer 以备后用。
   * @param port 要发送到 ApplciationMaster 的端口
   * @return 端口的序列化形式。
   */
  public static ByteBuffer serializeMetaData(int port) throws IOException {
    //这些字节应该被版本化
    DataOutputBuffer port_dob = new DataOutputBuffer();
    port_dob.writeInt(port);
    return ByteBuffer.wrap(port_dob.getData(), 0, port_dob.getLength());
  }

  /**
   * 用于反序列化 ShuffleHandler 返回的元数据的辅助函数。
   * @param meta -ShuffleHandler 返回的元数据
   * @return Shuffle Handler 监听的端口以提供 shuffle 数据。
   */
  public static int deserializeMetaData(ByteBuffer meta) throws IOException {
    //TODO this should be returning a class not just an int
    DataInputByteBuffer in = new DataInputByteBuffer();
    in.reset(meta);
    int port = in.readInt();
    return port;
  }

  /**
   * 一个辅助函数，用于序列化 JobTokenIdentifier 作为 ServiceData 发送到 ShuffleHandler。
   * @param jobToken 用于身份验证 Shuffle 数据请求的工作令牌。
   * @return jobToken 的序列化版本。
   */
  public static ByteBuffer serializeServiceData(Token<JobTokenIdentifier> jobToken) throws IOException {
    //TODO these bytes should be versioned
    DataOutputBuffer jobToken_dob = new DataOutputBuffer();
    jobToken.write(jobToken_dob);
    return ByteBuffer.wrap(jobToken_dob.getData(), 0, jobToken_dob.getLength());
  }

  //用于反序列化
  static Token<JobTokenIdentifier> deserializeServiceData(ByteBuffer secret) throws IOException {
    DataInputByteBuffer in = new DataInputByteBuffer();
    in.reset(secret);
    Token<JobTokenIdentifier> jt = new Token<JobTokenIdentifier>();
    jt.readFields(in);
    return jt;
  }

  @Override
  public void initializeApplication(ApplicationInitializationContext context) {//初始化应用程序，获取新的jobid并将其记录

    String user = context.getUser();
    ApplicationId appId = context.getApplicationId();
    ByteBuffer secret = context.getApplicationDataForService();
    // TODO these bytes should be versioned
    try {
      Token<JobTokenIdentifier> jt = deserializeServiceData(secret);//反序列化服务
       // TODO: Once SHuffle is out of NM, this can use MR APIs
      JobID jobId = new JobID(Long.toString(appId.getClusterTimestamp()), appId.getId());
      recordJobShuffleInfo(jobId, user, jt);//记录JobShuffle信息
    } catch (IOException e) {
      LOG.error("Error during initApp", e);
      // TODO add API to AuxiliaryServices to report failures
    }
  }

  @Override
  public void stopApplication(ApplicationTerminationContext context) {//终止应用程序
    ApplicationId appId = context.getApplicationId();
    JobID jobId = new JobID(Long.toString(appId.getClusterTimestamp()), appId.getId());
    try {
      removeJobShuffleInfo(jobId);//删除 JobShuffle 信息
    } catch (IOException e) {
      LOG.error("Error during stopApp", e);
      // TODO add API to AuxiliaryServices to report failures
    }
  }

  @Override
  protected void serviceInit(Configuration conf) throws Exception {//服务初始化
    manageOsCache = conf.getBoolean(SHUFFLE_MANAGE_OS_CACHE,
        DEFAULT_SHUFFLE_MANAGE_OS_CACHE);

    readaheadLength = conf.getInt(SHUFFLE_READAHEAD_BYTES,
        DEFAULT_SHUFFLE_READAHEAD_BYTES);
    
    maxShuffleConnections = conf.getInt(MAX_SHUFFLE_CONNECTIONS, 
                                        DEFAULT_MAX_SHUFFLE_CONNECTIONS);
    int maxShuffleThreads = conf.getInt(MAX_SHUFFLE_THREADS,
                                        DEFAULT_MAX_SHUFFLE_THREADS);
    if (maxShuffleThreads == 0) {
      maxShuffleThreads = 2 * Runtime.getRuntime().availableProcessors();
    }
    
    shuffleBufferSize = conf.getInt(SHUFFLE_BUFFER_SIZE, 
                                    DEFAULT_SHUFFLE_BUFFER_SIZE);
        
    shuffleTransferToAllowed = conf.getBoolean(SHUFFLE_TRANSFERTO_ALLOWED,
         (Shell.WINDOWS)?WINDOWS_DEFAULT_SHUFFLE_TRANSFERTO_ALLOWED:
                         DEFAULT_SHUFFLE_TRANSFERTO_ALLOWED);

    maxSessionOpenFiles = conf.getInt(SHUFFLE_MAX_SESSION_OPEN_FILES,
        DEFAULT_SHUFFLE_MAX_SESSION_OPEN_FILES);//最大会话打开文件

    ThreadFactory bossFactory = new ThreadFactoryBuilder()
      .setNameFormat("ShuffleHandler Netty Boss #%d")
      .build();
    ThreadFactory workerFactory = new ThreadFactoryBuilder()
      .setNameFormat("ShuffleHandler Netty Worker #%d")
      .build();
    
    selector = new NioServerSocketChannelFactory(
        HadoopExecutors.newCachedThreadPool(bossFactory),
        HadoopExecutors.newCachedThreadPool(workerFactory),
        maxShuffleThreads);
    super.serviceInit(new Configuration(conf));
  }

  // TODO change AbstractService to throw InterruptedException
  @Override
  protected void serviceStart() throws Exception {
    Configuration conf = getConfig();
    userRsrc = new ConcurrentHashMap<String,String>();//用户资源
    secretManager = new JobTokenSecretManager();
    recoverState(conf);
    ServerBootstrap bootstrap = new ServerBootstrap(selector);
    // Timer is shared across entire factory and must be released separately
    timer = new HashedWheelTimer();
    try {
      pipelineFact = new HttpPipelineFactory(conf, timer);
    } catch (Exception ex) {
      throw new RuntimeException(ex);
    }
    bootstrap.setOption("backlog", conf.getInt(SHUFFLE_LISTEN_QUEUE_SIZE,
        DEFAULT_SHUFFLE_LISTEN_QUEUE_SIZE));//shuffle监听队列大小
    bootstrap.setOption("child.keepAlive", true);
    bootstrap.setPipelineFactory(pipelineFact);
    port = conf.getInt(SHUFFLE_PORT_CONFIG_KEY, DEFAULT_SHUFFLE_PORT);//shuffle端口配置key
    Channel ch = bootstrap.bind(new InetSocketAddress(port));
    accepted.add(ch);//将shuffle端口添加至channel集群
    port = ((InetSocketAddress)ch.getLocalAddress()).getPort();
    conf.set(SHUFFLE_PORT_CONFIG_KEY, Integer.toString(port));
    pipelineFact.SHUFFLE.setPort(port);//在HttpPipelineFactory建立shuffle端口
    LOG.info(getName() + " listening on port " + port);
    super.serviceStart();

    sslFileBufferSize = conf.getInt(SUFFLE_SSL_FILE_BUFFER_SIZE_KEY,
                                    DEFAULT_SUFFLE_SSL_FILE_BUFFER_SIZE);//shuffle SSL 文件缓冲区大小key
    connectionKeepAliveEnabled =
        conf.getBoolean(SHUFFLE_CONNECTION_KEEP_ALIVE_ENABLED,
          DEFAULT_SHUFFLE_CONNECTION_KEEP_ALIVE_ENABLED);//是否对shuffle连接保持alive
    connectionKeepAliveTimeOut =
        Math.max(1, conf.getInt(SHUFFLE_CONNECTION_KEEP_ALIVE_TIME_OUT,
          DEFAULT_SHUFFLE_CONNECTION_KEEP_ALIVE_TIME_OUT));//shuffle连接保持alive超时时间
    mapOutputMetaInfoCacheSize =
        Math.max(1, conf.getInt(SHUFFLE_MAPOUTPUT_META_INFO_CACHE_SIZE,
          DEFAULT_SHUFFLE_MAPOUTPUT_META_INFO_CACHE_SIZE));//shuffle映射输出元信息缓存大小
  }

  @Override
  protected void serviceStop() throws Exception {
    accepted.close().awaitUninterruptibly(10, TimeUnit.SECONDS);
    if (selector != null) {//如果channelFactory不为空，则释放外部资源
      ServerBootstrap bootstrap = new ServerBootstrap(selector);//服务器引导程序
      bootstrap.releaseExternalResources();
    }
    if (pipelineFact != null) {//如果HttpPipelineFactory不为空则调用destroy()
      pipelineFact.destroy();
    }
    if (timer != null) {
      //释放这个共享定时器资源
      timer.stop();
    }
    if (stateDb != null) {
      stateDb.close();
    }
    ms.unregisterSource(ShuffleMetrics.class.getSimpleName());//取消注册源
    super.serviceStop();
  }

  @Override
  public synchronized ByteBuffer getMetaData() {
    try {
      return serializeMetaData(port); //序列化元数据
    } catch (IOException e) {
      LOG.error("Error during getMeta", e);
      // TODO add API to AuxiliaryServices to report failures
      return null;
    }
  }

  protected Shuffle getShuffle(Configuration conf) {
    return new Shuffle(conf);
  }

  private void recoverState(Configuration conf) throws IOException {//恢复状态
    Path recoveryRoot = getRecoveryPath();//获取恢复路径
    if (recoveryRoot != null) {
      startStore(recoveryRoot);
      Pattern jobPattern = Pattern.compile(JobID.JOBID_REGEX);//编译.JOBID_REGEX =JOB + SEPARATOR + "[0-9]+" + SEPARATOR + "[0-9]+";
      LeveldbIterator iter = null;//级别数据库迭代器
      try {
        iter = new LeveldbIterator(stateDb);
        iter.seek(bytes(JobID.JOB));
        while (iter.hasNext()) {
          Map.Entry<byte[],byte[]> entry = iter.next();
          String key = asString(entry.getKey());
          if (!jobPattern.matcher(key).matches()) {//将当前的key值与状态数据库进行匹配
            break;
          }
          recoverJobShuffleInfo(key, entry.getValue());
        }
      } catch (DBException e) {
        throw new IOException("Database error during recovery", e);
      } finally {
        if (iter != null) {
          iter.close();
        }
      }
    }
  }

  private void startStore(Path recoveryRoot) throws IOException {//打开数据库文件检查其版本，若本地数据库不存在则创建新的数据库
    Options options = new Options();
    options.createIfMissing(false);//若缺失则创建
    Path dbPath = new Path(recoveryRoot, STATE_DB_NAME);//数据库路径
    LOG.info("Using state database at " + dbPath + " for recovery");
    File dbfile = new File(dbPath.toString());
    try {
      stateDb = JniDBFactory.factory.open(dbfile, options);
    } catch (NativeDB.DBException e) {//若本地数据库不存在
      if (e.isNotFound() || e.getMessage().contains(" does not exist ")) {
        LOG.info("Creating state database at " + dbfile);
        options.createIfMissing(true);
        try {
          stateDb = JniDBFactory.factory.open(dbfile, options);
          storeVersion();
        } catch (DBException dbExc) {
          throw new IOException("Unable to create state store", dbExc);
        }
      } else {
        throw e;
      }
    }
    checkVersion();
  }
  
  @VisibleForTesting
  Version loadVersion() throws IOException {//获取当前版本
    byte[] data = stateDb.get(bytes(STATE_DB_SCHEMA_VERSION_KEY));
    // 如果先前未存储版本，则将其视为当前版本信息。
    if (data == null || data.length == 0) {
      return getCurrentVersion();
    }
    Version version =
        new VersionPBImpl(VersionProto.parseFrom(data));
    return version;
  }

  private void storeSchemaVersion(Version version) throws IOException {//将状态架构数据库的版本KEY值以及版本信息存入状态数据库中
    String key = STATE_DB_SCHEMA_VERSION_KEY;
    byte[] data = 
        ((VersionPBImpl) version).getProto().toByteArray();
    try {
      stateDb.put(bytes(key), data);
    } catch (DBException e) {
      throw new IOException(e.getMessage(), e);
    }
  }
  
  private void storeVersion() throws IOException {
    storeSchemaVersion(CURRENT_VERSION_INFO);
  }
  
  // 仅用于测试
  @VisibleForTesting
  void storeVersion(Version version) throws IOException {
    storeSchemaVersion(version);
  }

  protected Version getCurrentVersion() {
    return CURRENT_VERSION_INFO;
  }
  
  /**
   * 1) 版本方案：major.minor。 例如 1.0、1.1、1.2...1.25、2.0 等
   * 2) DB schema 的任何不兼容更改都是重大升级，DB schema 的任何兼容更改都是次要升级。
   * 3) 在小升级中，比如 1.1 到 1.2：覆盖版本信息并照常进行。
   * 4) 在重大升级中，比如 1.2 到 2.0:throw 异常并指示用户使用单独的升级工具来升级 shuffle 信息或删除不兼容的旧状态。
   */
  private void checkVersion() throws IOException {//检查加载版本与当前版本是否兼容，若兼容则调用storeVersion()方法存储该版本信息
    Version loadedVersion = loadVersion();
    LOG.info("Loaded state DB schema version info " + loadedVersion);
    if (loadedVersion.equals(getCurrentVersion())) {
      return;
    }
    if (loadedVersion.isCompatibleTo(getCurrentVersion())) {
      LOG.info("Storing state DB schema version info " + getCurrentVersion());
      storeVersion();
    } else {
      throw new IOException(
        "Incompatible version for state DB schema: expecting DB schema version " 
            + getCurrentVersion() + ", but loading version " + loadedVersion);
    }
  }

  private void addJobToken(JobID jobId, String user,
      Token<JobTokenIdentifier> jobToken) {//添加一个用户和jobId至用户资源，并为该jobId添加一个jobToken（jobToken 用于身份验证 Shuffle 数据请求的工作令牌。）
    userRsrc.put(jobId.toString(), user);
    secretManager.addTokenForJob(jobId.toString(), jobToken);
    LOG.info("Added token for " + jobId.toString());
  }

  private void recoverJobShuffleInfo(String jobIdStr, byte[] data)
      throws IOException {//对获取的jobId进行判断，若可用则添加至用户资源中
    JobID jobId;
    try {
      jobId = JobID.forName(jobIdStr);
    } catch (IllegalArgumentException e) {
      throw new IOException("Bad job ID " + jobIdStr + " in state store", e);
    }

    JobShuffleInfoProto proto = JobShuffleInfoProto.parseFrom(data);
    String user = proto.getUser();
    TokenProto tokenProto = proto.getJobToken();
    Token<JobTokenIdentifier> jobToken = new Token<JobTokenIdentifier>(
        tokenProto.getIdentifier().toByteArray(),
        tokenProto.getPassword().toByteArray(),
        new Text(tokenProto.getKind()), new Text(tokenProto.getService()));
    addJobToken(jobId, user, jobToken);
  }

  private void recordJobShuffleInfo(JobID jobId, String user,
      Token<JobTokenIdentifier> jobToken) throws IOException {
    if (stateDb != null) {//记录jobId的信息，并将其添加至状态数据库
      TokenProto tokenProto = TokenProto.newBuilder()
          .setIdentifier(ByteString.copyFrom(jobToken.getIdentifier()))
          .setPassword(ByteString.copyFrom(jobToken.getPassword()))
          .setKind(jobToken.getKind().toString())
          .setService(jobToken.getService().toString())
          .build();
      JobShuffleInfoProto proto = JobShuffleInfoProto.newBuilder()
          .setUser(user).setJobToken(tokenProto).build();
      try {
        stateDb.put(bytes(jobId.toString()), proto.toByteArray());
      } catch (DBException e) {
        throw new IOException("Error storing " + jobId, e);
      }
    }
    addJobToken(jobId, user, jobToken);
  }

  private void removeJobShuffleInfo(JobID jobId) throws IOException {//在用户资源中删除该jobId，并在状态数据库中删除jobId
    String jobIdStr = jobId.toString();
    secretManager.removeTokenForJob(jobIdStr);
    userRsrc.remove(jobIdStr);
    if (stateDb != null) {
      try {
        stateDb.delete(bytes(jobIdStr));
      } catch (DBException e) {
        throw new IOException("Unable to remove " + jobId
            + " from state store", e);
      }
    }
  }

  //超时处理程序
  static class TimeoutHandler extends IdleStateAwareChannelHandler {

    private boolean enabledTimeout;

    void setEnabledTimeout(boolean enabledTimeout) {
      this.enabledTimeout = enabledTimeout;
    }

    @Override
    public void channelIdle(ChannelHandlerContext ctx, IdleStateEvent e) {//如果超时则关闭连接
      if (e.getState() == IdleState.WRITER_IDLE && enabledTimeout) {
        e.getChannel().close();
      }
    }
  }

  class HttpPipelineFactory implements ChannelPipelineFactory {

    final Shuffle SHUFFLE;
    private SSLFactory sslFactory;
    private final ChannelHandler idleStateHandler;

    public HttpPipelineFactory(Configuration conf, Timer timer) throws Exception {
      SHUFFLE = getShuffle(conf);
      if (conf.getBoolean(MRConfig.SHUFFLE_SSL_ENABLED_KEY,
                          MRConfig.SHUFFLE_SSL_ENABLED_DEFAULT)) {
        LOG.info("Encrypted shuffle is enabled.");
        //创建一个sslFactory并添加相关配置和初始化
        sslFactory = new SSLFactory(SSLFactory.Mode.SERVER, conf);
        sslFactory.init();
      }
      //设置闲置状态处理程序
      this.idleStateHandler = new IdleStateHandler(timer, 0, connectionKeepAliveTimeOut, 0);

    }

    public Shuffle getSHUFFLE() {
      return SHUFFLE;
    }

    public void destroy() {
      if (sslFactory != null) {
        sslFactory.destroy();
      }
    }

    @Override
    public ChannelPipeline getPipeline() throws Exception {
      //将安全管理器纳入pipeline
      //排除编码/解码,以允许二进制混洗
      //分解索引解码以允许 alt.models
      ChannelPipeline pipeline = Channels.pipeline();
      if (sslFactory != null) {
        //添加SSL引擎到pipeline
        pipeline.addLast("ssl", new SslHandler(sslFactory.createSSLEngine()));
      }
      //添加Http请求解码器到pipeline
      pipeline.addLast("decoder", new HttpRequestDecoder());
      //添加Http块聚合器到pipeline
      pipeline.addLast("aggregator", new HttpChunkAggregator(1 << 16));
      //添加Http响应编码器到pipeline
      pipeline.addLast("encoder", new HttpResponseEncoder());
      //添加块写入处理程序到pipeline
      pipeline.addLast("chunking", new ChunkedWriteHandler());
      //添加shuffle到pipeline
      pipeline.addLast("shuffle", SHUFFLE);
      //添加空闲状态处理程序到pipeline
      pipeline.addLast("idle", idleStateHandler);
      //添加超时处理程序到pipeline
      pipeline.addLast(TIMEOUT_HANDLER, new TimeoutHandler());
      return pipeline;
    }
  }

  class Shuffle extends SimpleChannelUpstreamHandler {
    private final IndexCache indexCache;//索引缓存
    private final
    LoadingCache<AttemptPathIdentifier, AttemptPathInfo> pathCache;//路径缓存

    private int port;

    Shuffle(Configuration conf) {//shuffle配置
      this.port = conf.getInt(SHUFFLE_PORT_CONFIG_KEY, DEFAULT_SHUFFLE_PORT);//shuffle端口key
      this.indexCache = new IndexCache(new JobConf(conf));
      this.pathCache = CacheBuilder.newBuilder()//缓存生成器
          .expireAfterAccess(conf.getInt(EXPIRE_AFTER_ACCESS_MINUTES,
              DEFAULT_EXPIRE_AFTER_ACCESS_MINUTES), TimeUnit.MINUTES)//访问后过期
          .softValues()
          .concurrencyLevel(conf.getInt(CONCURRENCY_LEVEL,
              DEFAULT_CONCURRENCY_LEVEL))//并发级别
          .removalListener((RemovalListener<AttemptPathIdentifier,
              AttemptPathInfo>) notification ->
              LOG.debug("PathCache Eviction: {}, Reason={}",
                  notification.getKey(), notification.getCause()))
          .maximumWeight(conf.getInt(MAX_WEIGHT, DEFAULT_MAX_WEIGHT))//最大路径缓存
          .weigher((key, value) -> key.jobId.length() + key.user.length() +
              key.attemptId.length()+ value.indexPath.toString().length() +
              value.dataPath.toString().length())
          .build(new CacheLoader<AttemptPathIdentifier, AttemptPathInfo>() {//缓存加载器
            @Override
            public AttemptPathInfo load(AttemptPathIdentifier key) throws
                Exception {
              String base = getBaseLocation(key.jobId, key.user);//获取基本位置
              String attemptBase = base + key.attemptId;
              Path indexFileName = getAuxiliaryLocalPathHandler()//获取辅助本地路径处理程序
                  .getLocalPathForRead(attemptBase + "/" + INDEX_FILE_NAME);//获取本地路径读取
              Path mapOutputFileName = getAuxiliaryLocalPathHandler()
                  .getLocalPathForRead(attemptBase + "/" + DATA_FILE_NAME);
              LOG.debug("Loaded : {} via loader", key);
              return new AttemptPathInfo(indexFileName, mapOutputFileName);
            }
          });
    }

    public void setPort(int port) {
      this.port = port;
    }

    private List<String> splitMaps(List<String> mapq) {//拆分map
      if (null == mapq) {
        return null;
      }
      final List<String> ret = new ArrayList<String>();
      for (String s : mapq) {
        Collections.addAll(ret, s.split(","));
      }
      return ret;
    }

    @Override
    public void channelOpen(ChannelHandlerContext ctx, ChannelStateEvent evt) 
        throws Exception {
      super.channelOpen(ctx, evt);//把通道状态事件发送给上游

      //当前 shuffle 连接数 大于或等于最大允许 shuffle 连接数
      if ((maxShuffleConnections > 0) && (accepted.size() >= maxShuffleConnections)) {
        LOG.info(String.format("Current number of shuffle connections (%d) is " + 
            "greater than or equal to the max allowed shuffle connections (%d)", 
            accepted.size(), maxShuffleConnections));

        Map<String, String> headers = new HashMap<String, String>(1);
        // 如果达到 shuffle 连接限制，则通知 fetchers 在关闭连接之前退避一段时间。
        // fetcher 应该优雅地处理这个通知，也就是说，不把它当作 fetch 失败。
        headers.put(RETRY_AFTER_HEADER, String.valueOf(FETCH_RETRY_DELAY));
        sendError(ctx, "", TOO_MANY_REQ_STATUS, headers);
        return;
      }
      accepted.add(evt.getChannel());//将此时通道状态事件的通道添加到channel集群中
    }

    @Override
    public void messageReceived(ChannelHandlerContext ctx, MessageEvent evt)
        throws Exception {
      HttpRequest request = (HttpRequest) evt.getMessage();
      if (request.getMethod() != GET) {
          sendError(ctx, METHOD_NOT_ALLOWED);//方法不被允许
          return;
      }
      // 检查shuffle版本是否兼容
      if (!ShuffleHeader.DEFAULT_HTTP_HEADER_NAME.equals(
          request.headers() != null ?
              request.headers().get(ShuffleHeader.HTTP_HEADER_NAME) : null)
          || !ShuffleHeader.DEFAULT_HTTP_HEADER_VERSION.equals(
              request.headers() != null ?
                  request.headers()
                      .get(ShuffleHeader.HTTP_HEADER_VERSION) : null)) {
        sendError(ctx, "Incompatible shuffle request version", BAD_REQUEST);//不兼容的 shuffle 请求版本
      }
      final Map<String,List<String>> q =
        new QueryStringDecoder(request.getUri()).getParameters();//查询字符串解码器,将uri转换为参数
      final List<String> keepAliveList = q.get("keepAlive");//key：keepAlive映射到的值
      boolean keepAliveParam = false;
      if (keepAliveList != null && keepAliveList.size() == 1) {
        keepAliveParam = Boolean.valueOf(keepAliveList.get(0));//返回index:0对应的值的布尔量
        if (LOG.isDebugEnabled()) {//已启用调试
          LOG.debug("KeepAliveParam : " + keepAliveList
            + " : " + keepAliveParam);
        }
      }
      final List<String> mapIds = splitMaps(q.get("map"));
      final List<String> reduceQ = q.get("reduce");
      final List<String> jobQ = q.get("job");
      if (LOG.isDebugEnabled()) {
        LOG.debug("RECV: " + request.getUri() +
            "\n  mapId: " + mapIds +
            "\n  reduceId: " + reduceQ +
            "\n  jobId: " + jobQ +
            "\n  keepAlive: " + keepAliveParam);
      }

      if (mapIds == null || reduceQ == null || jobQ == null) {
        sendError(ctx, "Required param job, map and reduce", BAD_REQUEST);
        return;
      }
      if (reduceQ.size() != 1 || jobQ.size() != 1) {
        sendError(ctx, "Too many job/reduce parameters", BAD_REQUEST);
        return;
      }

      int reduceId;
      String jobId;
      try {
        reduceId = Integer.parseInt(reduceQ.get(0));
        jobId = jobQ.get(0);
      } catch (NumberFormatException e) {
        sendError(ctx, "Bad reduce parameter", BAD_REQUEST);
        return;
      } catch (IllegalArgumentException e) {
        sendError(ctx, "Bad job parameter", BAD_REQUEST);
        return;
      }
      final String reqUri = request.getUri();
      if (null == reqUri) {//uri是除去host（域名或者ip）部分的路径
        // TODO? add upstream?
        sendError(ctx, FORBIDDEN);
        return;
      }
      HttpResponse response = new DefaultHttpResponse(HTTP_1_1, OK);//HTTP版本号为：HTTP_1_1，状态为：OK
      try {
        verifyRequest(jobId, ctx, request, response,
            new URL("http", "", this.port, reqUri));//验证http请求
      } catch (IOException e) {
        LOG.warn("Shuffle failure ", e);
        sendError(ctx, e.getMessage(), UNAUTHORIZED);
        return;
      }

      Map<String, MapOutputInfo> mapOutputInfoMap =
          new HashMap<String, MapOutputInfo>();
      Channel ch = evt.getChannel();
      ChannelPipeline pipeline = ch.getPipeline();
      TimeoutHandler timeoutHandler =
          (TimeoutHandler)pipeline.get(TIMEOUT_HANDLER);
      timeoutHandler.setEnabledTimeout(false);
      String user = userRsrc.get(jobId);

      try {
        populateHeaders(mapIds, jobId, user, reduceId, request,
          response, keepAliveParam, mapOutputInfoMap);//填充标题
      } catch(IOException e) {
        ch.write(response);
        LOG.error("Shuffle error in populating headers :", e);//填充标题中的shuffle错误：
        String errorMessage = getErrorMessage(e);
        sendError(ctx,errorMessage , INTERNAL_SERVER_ERROR);
        return;
      }
      ch.write(response);//将响应写入事件通道中
      //每个 messageReceived 调用初始化一个 ReduceContext 对象
      boolean keepAlive = keepAliveParam || connectionKeepAliveEnabled;
      ReduceContext reduceContext = new ReduceContext(mapIds, reduceId, ctx,
          user, mapOutputInfoMap, jobId, keepAlive);
      for (int i = 0; i < Math.min(maxSessionOpenFiles, mapIds.size()); i++) {
        ChannelFuture nextMap = sendMap(reduceContext);
        if(nextMap == null) {
          return;
        }
      }
    }

    /**
     * 为 ReduceContext.mapsToSend 指向的 mapId 调用 sendMapOutput 并增加它。
     * 此方法首先由 messageReceived() maxSessionOpenFiles 次调用，然后在每个 sendMapOutput 操作完成时调用。
     * 这限制了节点上打开文件的数量，如果一次性调用所有 sendMapOutputs，这会变得非常大（耗尽 NM 上的文件描述符），就像在此更改之前所做的那样。
     * @param reduceContext 用于使用正确的参数调用 sendMapOutput。
     * @return sendMapOutput 的ChannelFuture，可以为null。
     */
    public ChannelFuture sendMap(ReduceContext reduceContext)
        throws Exception {//获取map输出的分区或者分块
      ChannelFuture nextMap = null;
      if (reduceContext.getMapsToSend().get() < reduceContext.getMapIds().size()) {
        int nextIndex = reduceContext.getMapsToSend().getAndIncrement();
        String mapId = reduceContext.getMapIds().get(nextIndex);

        try {
          MapOutputInfo info = reduceContext.getInfoMap().get(mapId);
          if (info == null) {
            info = getMapOutputInfo(mapId, reduceContext.getReduceId(),
                reduceContext.getJobId(), reduceContext.getUser());
          }
          nextMap = sendMapOutput(
              reduceContext.getCtx(),
              reduceContext.getCtx().getChannel(),
              reduceContext.getUser(), mapId,
              reduceContext.getReduceId(), info);
          if (null == nextMap) {
            sendError(reduceContext.getCtx(), NOT_FOUND);
            return null;
          }
          nextMap.addListener(new ReduceMapFileCount(reduceContext));//添加一个监听器
        } catch (IOException e) {
          if (e instanceof DiskChecker.DiskErrorException) {
            LOG.error("Shuffle error :" + e);
          } else {
            LOG.error("Shuffle error :", e);
          }
          String errorMessage = getErrorMessage(e);
          sendError(reduceContext.getCtx(), errorMessage,
              INTERNAL_SERVER_ERROR);
          return null;
        }
      }
      return nextMap;
    }

    private String getErrorMessage(Throwable t) {
      //获取错误信息
      StringBuffer sb = new StringBuffer(t.getMessage());
      while (t.getCause() != null) {
        sb.append(t.getCause().getMessage());
        t = t.getCause();
      }
      return sb.toString();
    }

    private String getBaseLocation(String jobId, String user) {
      //获取用户程序的基本位置，其中包含用户缓存，应用程序缓存以及应用程序的id
      final JobID jobID = JobID.forName(jobId);
      final ApplicationId appID =
          ApplicationId.newInstance(Long.parseLong(jobID.getJtIdentifier()),
            jobID.getId());
      final String baseStr =
          ContainerLocalizer.USERCACHE + "/" + user + "/"
              + ContainerLocalizer.APPCACHE + "/"
              + appID.toString() + "/output" + "/";
      return baseStr;
    }

    protected MapOutputInfo getMapOutputInfo(String mapId, int reduce,
        String jobId, String user) throws IOException {//获取map输出的信息
      AttemptPathInfo pathInfo;
      try {
        AttemptPathIdentifier identifier = new AttemptPathIdentifier(
            jobId, user, mapId);//身份信息
        pathInfo = pathCache.get(identifier);//路径信息
        if (LOG.isDebugEnabled()) {
          LOG.debug("Retrieved pathInfo for " + identifier +
              " check for corresponding loaded messages to determine whether" +
              " it was loaded or cached");//检索到identifier的路径信息,检查相应的加载消息以确定它是否已加载或缓存
        }
      } catch (ExecutionException e) {
        if (e.getCause() instanceof IOException) {
          throw (IOException) e.getCause();
        } else {
          throw new RuntimeException(e.getCause());
        }
      }

      IndexRecord info =
        indexCache.getIndexInformation(mapId, reduce, pathInfo.indexPath, user);

      if (LOG.isDebugEnabled()) {
        LOG.debug("getMapOutputInfo: jobId=" + jobId + ", mapId=" + mapId +
            ",dataFile=" + pathInfo.dataPath + ", indexFile=" +
            pathInfo.indexPath);
      }

      MapOutputInfo outputInfo = new MapOutputInfo(pathInfo.dataPath, info);
      return outputInfo;
    }

    protected void populateHeaders(List<String> mapIds, String jobId,
        String user, int reduce, HttpRequest request, HttpResponse response,
        boolean keepAliveParam, Map<String, MapOutputInfo> mapOutputInfoMap)
        throws IOException {

      long contentLength = 0;
      for (String mapId : mapIds) {
        MapOutputInfo outputInfo = getMapOutputInfo(mapId, reduce, jobId, user);
        if (mapOutputInfoMap.size() < mapOutputMetaInfoCacheSize) {
          mapOutputInfoMap.put(mapId, outputInfo);
        }//将map输出信息放入mapOutputInfoMap

        ShuffleHeader header =
            new ShuffleHeader(mapId, outputInfo.indexRecord.partLength,
            outputInfo.indexRecord.rawLength, reduce);//partlength为数据压缩后的长度
        DataOutputBuffer dob = new DataOutputBuffer();//数据输出缓冲区
        header.write(dob);//将数据缓冲区该对象写入header

        contentLength += outputInfo.indexRecord.partLength;
        contentLength += dob.getLength();
      }

      // 现在设置响应头。
      setResponseHeaders(response, keepAliveParam, contentLength);

      // 默认情况下禁用此审核日志，
      // 要打开它，请启用此审核日志
      if (AUDITLOG.isDebugEnabled()) {
        StringBuilder sb = new StringBuilder("shuffle for ");
        sb.append(jobId).append(" reducer ").append(reduce);
        sb.append(" length ").append(contentLength);
        if (AUDITLOG.isTraceEnabled()) {
          // 对于跟踪级别的日志记录，附加映射器列表
          sb.append(" mappers: ").append(mapIds);
          AUDITLOG.trace(sb.toString());
        } else {
          AUDITLOG.debug(sb.toString());
        }
      }
    }

    //设置响应头
    protected void setResponseHeaders(HttpResponse response,
        boolean keepAliveParam, long contentLength) {
      if (!connectionKeepAliveEnabled && !keepAliveParam) {//不能够保持连接alive且keepAliveParam为false
        if (LOG.isDebugEnabled()) {
          LOG.debug("Setting connection close header...");//设置连接关闭头
        }
        response.headers().set(HttpHeader.CONNECTION.asString(),
            CONNECTION_CLOSE);//设置连接关闭
      } else {
        response.headers().set(HttpHeader.CONTENT_LENGTH.asString(),
          String.valueOf(contentLength));//内容长度
        response.headers().set(HttpHeader.CONNECTION.asString(),
            HttpHeader.KEEP_ALIVE.asString());
        response.headers().set(HttpHeader.KEEP_ALIVE.asString(),
            "timeout=" + connectionKeepAliveTimeOut);//keepalive的超时时间
        LOG.info("Content Length in shuffle : " + contentLength);
      }
    }

    //获取map端输出的文件名以及索引记录
    class MapOutputInfo {
      final Path mapOutputFileName;
      final IndexRecord indexRecord;

      MapOutputInfo(Path mapOutputFileName, IndexRecord indexRecord) {
        this.mapOutputFileName = mapOutputFileName;
        this.indexRecord = indexRecord;
      }
    }

    protected void verifyRequest(String appid, ChannelHandlerContext ctx,
        HttpRequest request, HttpResponse response, URL requestUri)
        throws IOException {//验证请求
      SecretKey tokenSecret = secretManager.retrieveTokenSecret(appid);//检索令牌
      if (null == tokenSecret) {
        LOG.info("Request for unknown token " + appid);
        throw new IOException("could not find jobid");
      }
      // 要加密的字符串
      String enc_str = SecureShuffleUtils.buildMsgFrom(requestUri);
      // hash from the fetcher
      String urlHashStr =
          request.headers().get(SecureShuffleUtils.HTTP_HEADER_URL_HASH);
      if (urlHashStr == null) {
        LOG.info("Missing header hash for " + appid);
        throw new IOException("fetcher cannot be authenticated");
      }
      if (LOG.isDebugEnabled()) {
        int len = urlHashStr.length();
        LOG.debug("verifying request. enc_str=" + enc_str + "; hash=..." +
            urlHashStr.substring(len-len/2, len-1));//截取urlHashStr的一半
      }
      // 验证 - 抛出异常
      SecureShuffleUtils.verifyReply(urlHashStr, enc_str, tokenSecret);
      // 验证通过 - 对回复进行编码
      String reply =
        SecureShuffleUtils.generateHash(urlHashStr.getBytes(Charsets.UTF_8), 
            tokenSecret);
      // 将 url的hash值，HTTP标头名字，HTTP标头版本放入 respone标头中
      response.headers().set(
          SecureShuffleUtils.HTTP_HEADER_REPLY_URL_HASH, reply);//url的hash值
      response.headers().set(ShuffleHeader.HTTP_HEADER_NAME,
          ShuffleHeader.DEFAULT_HTTP_HEADER_NAME);//HTTP标头名字
      response.headers().set(ShuffleHeader.HTTP_HEADER_VERSION,
          ShuffleHeader.DEFAULT_HTTP_HEADER_VERSION);//HTTP标头版本
      if (LOG.isDebugEnabled()) {
        int len = reply.length();
        LOG.debug("Fetcher request verfied. enc_str=" + enc_str + ";reply=" +
            reply.substring(len-len/2, len-1));
      }
    }

    protected ChannelFuture sendMapOutput(ChannelHandlerContext ctx, Channel ch,
        String user, String mapId, int reduce, MapOutputInfo mapOutputInfo)
        throws IOException {
      final IndexRecord info = mapOutputInfo.indexRecord;
      final ShuffleHeader header =
        new ShuffleHeader(mapId, info.partLength, info.rawLength, reduce);
      final DataOutputBuffer dob = new DataOutputBuffer();//数据输出缓冲区
      header.write(dob);
      ch.write(wrappedBuffer(dob.getData(), 0, dob.getLength()));
      final File spillfile =
          new File(mapOutputInfo.mapOutputFileName.toString());//map输出文件
      RandomAccessFile spill;
      try {
        spill = SecureIOUtils.openForRandomRead(spillfile, "r", user, null);
      } catch (FileNotFoundException e) {
        LOG.info(spillfile + " not found");
        return null;
      }
      ChannelFuture writeFuture;
      if (ch.getPipeline().get(SslHandler.class) == null) {//如果通道为空则允许传输
        final FadvisedFileRegion partition = new FadvisedFileRegion(spill,
            info.startOffset, info.partLength, manageOsCache, readaheadLength,
            readaheadPool, spillfile.getAbsolutePath(), 
            shuffleBufferSize, shuffleTransferToAllowed);
        writeFuture = ch.write(partition);
        writeFuture.addListener(new ChannelFutureListener() {
            // TODO error handling; distinguish IO/connection failures,
            //      attribute to appropriate spill output
          @Override
          public void operationComplete(ChannelFuture future) {
            if (future.isSuccess()) {
              partition.transferSuccessful();
            }
            partition.releaseExternalResources();
          }
        });
      } else {
        // HTTPS 不能用零拷贝来完成。
        final FadvisedChunkedFile chunk = new FadvisedChunkedFile(spill,
            info.startOffset, info.partLength, sslFileBufferSize,
            manageOsCache, readaheadLength, readaheadPool,
            spillfile.getAbsolutePath());
        writeFuture = ch.write(chunk);
      }
      metrics.shuffleConnections.incr();
      metrics.shuffleOutputBytes.incr(info.partLength); //对数据压缩后的长度进行自增
      return writeFuture;
    }

    protected void sendError(ChannelHandlerContext ctx,
        HttpResponseStatus status) {
      sendError(ctx, "", status);
    }

    protected void sendError(ChannelHandlerContext ctx, String message,
        HttpResponseStatus status) {
      sendError(ctx, message, status, Collections.<String, String>emptyMap());
    }

    protected void sendError(ChannelHandlerContext ctx, String msg,
        HttpResponseStatus status, Map<String, String> headers) {
      HttpResponse response = new DefaultHttpResponse(HTTP_1_1, status);
      response.headers().set(CONTENT_TYPE, "text/plain; charset=UTF-8");
      // 将 shuffle 版本放入 http 标头中
      response.headers().set(ShuffleHeader.HTTP_HEADER_NAME,
          ShuffleHeader.DEFAULT_HTTP_HEADER_NAME);
      response.headers().set(ShuffleHeader.HTTP_HEADER_VERSION,
          ShuffleHeader.DEFAULT_HTTP_HEADER_VERSION);
      for (Map.Entry<String, String> header : headers.entrySet()) {
        response.headers().set(header.getKey(), header.getValue());
      }
      response.setContent(
          ChannelBuffers.copiedBuffer(msg, CharsetUtil.UTF_8));

      //发送错误消息后立即关闭连接。
      ctx.getChannel().write(response).addListener(ChannelFutureListener.CLOSE);
    }

    @Override
    //异常捕获
    public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e)
        throws Exception {
      Channel ch = e.getChannel();
      Throwable cause = e.getCause();
      if (cause instanceof TooLongFrameException) {
        sendError(ctx, BAD_REQUEST);
        return;
      } else if (cause instanceof IOException) {
        if (cause instanceof ClosedChannelException) {
          LOG.debug("Ignoring closed channel error", cause);
          return;
        }
        String message = String.valueOf(cause.getMessage());
        if (IGNORABLE_ERROR_MESSAGE.matcher(message).matches()) {
          LOG.debug("Ignoring client socket close", cause);
          return;
        }
      }

      LOG.error("Shuffle error: ", cause);
      if (ch.isConnected()) {
        LOG.error("Shuffle error " + e);
        sendError(ctx, INTERNAL_SERVER_ERROR);
      }
    }
  }

  /**
   * 将其更改为仅存储本地目录索引，而不是整个路径。效率更高。
   */
  static class AttemptPathInfo {
    private final Path indexPath;
    private final Path dataPath;

    public AttemptPathInfo(Path indexPath, Path dataPath) {
      this.indexPath = indexPath;
      this.dataPath = dataPath;
    }
  }

  static class AttemptPathIdentifier {
    private final String jobId;
    private final String user;
    private final String attemptId;

    public AttemptPathIdentifier(String jobId, String user, String attemptId) {
      this.jobId = jobId;
      this.user = user;
      this.attemptId = attemptId;
    }

    @Override
    public boolean equals(Object o) {
      if (this == o) {
        return true;
      }
      if (o == null || getClass() != o.getClass()) {
        return false;
      }

      AttemptPathIdentifier that = (AttemptPathIdentifier) o;

      if (!attemptId.equals(that.attemptId)) {
        return false;
      }
      if (!jobId.equals(that.jobId)) {
        return false;
      }

      return true;
    }

    @Override
    public int hashCode() {
      int result = jobId.hashCode();
      result = 31 * result + attemptId.hashCode();
      return result;
    }

    @Override
    public String toString() {
      return "AttemptPathIdentifier{" +
          "attemptId='" + attemptId + '\'' +
          ", jobId='" + jobId + '\'' +
          '}';
    }
  }
}
